package com.atguigu.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author Felix
 * @date 2024/3/4
 * 该案例演示了FlinkSQL的环境准备以及从流中读取数据转换为动态表
 *      流--->动态表--->持续查询--->动态表--->流
 */
public class Flink01_Sql_Env {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(1);
        //1.3 指定表执行环境
        //StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, EnvironmentSettings.newInstance()
        //        .inStreamingMode()
        //        .build());
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //TODO 2.从指定的流中读取数据  创建动态表---通过连接器创建表
        tableEnv.executeSql("CREATE TABLE source ( \n" +
                "    id INT, \n" +
                "    ts BIGINT, \n" +
                "    vc INT \n" +
                ") WITH ( \n" +
                "    'connector' = 'datagen', \n" +
                "    'rows-per-second'='1', \n" +
                "    'fields.id.kind'='random', \n" +
                "    'fields.id.min'='1', \n" +
                "    'fields.id.max'='10', \n" +
                "    'fields.ts.kind'='sequence', \n" +
                "    'fields.ts.start'='1', \n" +
                "    'fields.ts.end'='1000000', \n" +
                "    'fields.vc.kind'='random', \n" +
                "    'fields.vc.min'='1', \n" +
                "    'fields.vc.max'='100'\n" +
                ")");


        //通过虚拟视图的方式创建表
        //DataStreamSource<Integer> ds1 = env.fromElements(1, 2, 3, 4);
        //tableEnv.createTemporaryView("test",ds1);

        //为什么叫虚拟视图呢？
        //执行查询操作方式1   这种方式参数除了可以接受DQL之外，还可以接收其它操作 ，返回的直接就是查询结果
        //TableResult tableResult1 = tableEnv.executeSql("select * from source");
        //tableResult1.print();
        //执行查询操作方式2   参数只能接收DQL，返回结果是动态表对象
        //Table resTable = tableEnv.sqlQuery("select * from source");
        //table.execute().print();


        //这种写法报错   Object 'resTable' not found
        //tableEnv.executeSql("select * from resTable").print();
        //这种方式底层  在TableImpl的toString方法中，将表对象注册到了表执行环境中
        //tableEnv.executeSql("select * from " + resTable).print();

        //tableEnv.createTemporaryView("res_table",resTable);
        //tableEnv.executeSql("select * from res_table").print();

        //通过TableAPI的方式从表中查询数据
        //Table sourceTable = tableEnv.sqlQuery("select * from source");
        //id ts  vc   select id,vc from biao where id=3;
        //Table selectTable = sourceTable
        //        .where($("id").isEqual(3))
        //        .select($("id"), $("vc"));
        //selectTable.execute().print();


        //TODO 3.从表中查询数据  并打印输出
        //TableResult tableResult = tableEnv.executeSql("select * from test");
        //tableResult.print();

        //TODO 4.创建输出表
        tableEnv.executeSql("CREATE TABLE sink (\n" +
                "    id INT, \n" +
                "    ts BIGINT, \n" +
                "    vc INT\n" +
                ") WITH (\n" +
                "'connector' = 'print'\n" +
                ")");

        //从指定的source表中查询数据，将查询结果通过sink表进行输出
        //tableEnv.executeSql("insert into sink select * from source");


        //在测试的时候，不要让程序中存在多个print,因为对于sql中的无界流，
        //tableEnv.executeSql("select id,ts,vc from source").print();
        //tableEnv.executeSql("select id,vc from source").print();



        //TODO 5. 提交  注意：如果程序最后操作的是SQL动态表，不需要显示通过env.execute提交

    }
}
